// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: leon

#include "coin2/exchange/bitflyer_futures/feed_v1/processor.h"

#include <string>
#include <utility>

#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/bitflyer_futures/feed_v1/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::bitflyer_futures::feed_v1::impl {

void BitflyerFuturesFeedProcessor::OnTopicRecord(const base::executor::TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  // Possible update messages:
  // 0{"sid":"7YQt2wdMyTmFyZJgACBH","upgrades":[],"pingInterval":15000,"pingTimeout":3000}
  // 3
  // 40
  // 42[\"lightning_board_FX_BTC_JPY\",{\"mid_price\":763796,\"bids\":[],\"asks\":[{\"price\":764037,\"size\":0.02}]}]"
  // 42[\"lightning_executions_FX_BTC_JPY\",[{\"id\":1685717883,\"side\":\"BUY\",\"price\":763796,\"size\":0.02,\"exec_date\":\"2020-04-16T09:35:49.8843855Z\",\"buy_child_order_acceptance_id\":\"JRF20200416-093549-575381\",\"sell_child_order_acceptance_id\":\"JRF20200416-093549-228488\"}]]
  const auto& data = packet.payload;
  // Heart
  if (data == "3" || data == "40" || data.substr(0, 2) == "0{") {
    return;
  }

  rapidjson::Document document;
  auto pos = data.find_first_of('[');
  if (pos == std::string::npos || pos >= 3) {
    LOG(INFO) << "Not target JSON message: " << packet.payload;
    return;
  }

  document.Parse<rapidjson::kParseNumbersAsStringsFlag>(data.c_str() + pos);
  if (document.HasParseError()) {
    THROW() << "JSON parse error: " << GetParseError_En(document.GetParseError()) << packet.payload;
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  if (document.IsArray()) {
    auto items = document.GetArray();
    CHECK_EQ_THROW(items.Size(), 2)
        << "[Bitflyer Futures Feed Processor] unexpected list length: " << packet.payload;

    const auto& channel = BitflyerFuturesChannel::Parse(items[0].GetString());

    if (!CheckChannelFilter(channel.channel)) {
      return;
    }

    const auto& native_symbol = channel.native_symbol;
    if (channel.channel == "lightning_executions") {
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        return;
      }
      Trade* agg_trade = upd_mgr_.GetTradeMutable();
      agg_trade->Clear();
      Trade trade{};

      const auto& trades = items[1].GetArray();
      agg_trade->timestamp = TimestampFromIso8601(trades[0]["exec_date"].GetString());

      for (auto& trade_json : trades) {
        BitflyerFuturesFeedUpdateParser::ParseTrade(trade_json, &trade);
        if (TryAggregateTrades(agg_trade, trade)) {
          continue;
        }
        if (agg_trade->fill_qty > 0) {
          upd_mgr_.PublishTrade(native_symbol, true);
        }
        *agg_trade = trade;
      }
      if (agg_trade->fill_qty > 0) {
        upd_mgr_.PublishTrade(native_symbol, false);
      }
    } else if (channel.channel == "lightning_board") {
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        if (book->BidSize() > 0 && book->AskSize() > 0) {
          bool book_valid =
              BitflyerFuturesFeedUpdateParser::ParseUpdate(book, items[1], packet.timestamp);
          if (book_valid) {
            upd_mgr_.PublishBook();
          }
        }
      }
    } else if (channel.channel == "lightning_board_snapshot") {
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        if (book->BidSize() == 0 || book->AskSize() == 0) {
          // Only handle snapshot if book is empty.
          // Due to Bitflyer having 10m maintenance period everyday,
          // book will be cleared and initialized with snapshot at that time.
          bool book_valid =
              BitflyerFuturesFeedUpdateParser::ParseSnapshot(book, items[1], packet.timestamp);
          if (book_valid) {
            upd_mgr_.PublishBook();
          } else {
            LOG(ERROR) << "[Bitflyer Futures Feed Processor] wrong snapshot: " << packet.topic_name
                       << ": " << packet.payload.data();
          }
        }
      }
    } else {
      LOG(INFO) << "[Bitflyer Futures Feed Processor] unknown msg for " << packet.topic_name << ": "
                << packet.payload.data();
    }
  } else {
    LOG(INFO) << "[Bitflyer Futures Feed Processor] meta msg for " << packet.topic_name << ": "
              << packet.payload.data();
  }

  upd_mgr_.onEndPacket(packet);
}  // namespace coin2::exchange::bitflyer_futures::feed_v1::impl
}  // namespace coin2::exchange::bitflyer_futures::feed_v1::impl
